package cn.getech.data.development.service.impl;

import cn.getech.data.development.constant.RunStatus;
import cn.getech.data.development.entity.flink.FlinkNodeField;
import cn.getech.data.development.entity.flink.FlinkNodeUnion;
import cn.getech.data.development.entity.flink.FlinkTaskNode;
import cn.getech.data.development.mapper.RealTimeTaskMapper;
import cn.getech.data.development.mapper.flink.FlinkNodeFieldMapper;
import cn.getech.data.development.mapper.flink.FlinkNodeUnionMapper;
import cn.getech.data.development.mapper.flink.FlinkTaskNodeMapper;
import cn.getech.data.development.model.vo.FlinkTaskNodeVO;
import cn.getech.data.development.model.vo.RealTimeTaskVO;
import cn.getech.data.development.service.FlinkTaskNodeService;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.system.center.utils.ShiroUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class FlinkTaskNodeServiceImpl extends ServiceImpl<FlinkTaskNodeMapper, FlinkTaskNode> implements FlinkTaskNodeService {

    @Autowired
    private FlinkTaskNodeMapper flinkTaskNodeMapper;
    @Autowired
    private FlinkNodeFieldMapper flinkNodeFieldMapper;
    @Autowired
    private FlinkNodeUnionMapper flinkNodeUnionMapper;
    @Autowired
    private RealTimeTaskMapper realTimeTaskMapper;
    @Override
    public void removeNode(FlinkTaskNodeVO taskNodeVO) {
        Integer id=taskNodeVO.getId();
        if (id != null) {
            flinkTaskNodeMapper.deleteById(id);
            flinkNodeUnionMapper.delete(new QueryWrapper<FlinkNodeUnion>().eq("node_from",id));
            flinkNodeUnionMapper.delete(new QueryWrapper<FlinkNodeUnion>().eq("node_to",id));
        }
        if (taskNodeVO.getConfig() != null) {
            String config = taskNodeVO.getConfig().toJSONString();
            realTimeTaskMapper.updateConfig(config, taskNodeVO.getTaskId());
        }
    }

    @Override
    public List<String> fieldType() {
        return flinkNodeFieldMapper.selectFieldType();
    }

    @Override
    public void saveNode(FlinkTaskNodeVO vo) {
        FlinkTaskNode taskNode = vo.toDto();
        Long userId = ShiroUtils.getUserId();
        RealTimeTaskVO task = realTimeTaskMapper.getByMenuId(vo.getTaskId());
        if (task != null) {
            if (RunStatus.RUNNING.getStatus().equals(task.getRunState())) {
                throw new RRException("任务运行中，节点不可修改");
            }
        }
        taskNode.preSave(userId);
        if (vo.getConfig() != null) {
            Integer sourceId = vo.getConfig().getInteger("sourceId");
            taskNode.setSourceId(sourceId);
        }
        if (taskNode.getId() != null) {
            flinkTaskNodeMapper.updateSelectiveById(taskNode);
        }else{
            FlinkTaskNode node = flinkTaskNodeMapper.selectOne(new QueryWrapper<FlinkTaskNode>()
                    .eq("node_name", vo.getNodeName()).eq("task_id",vo.getTaskId()));
            if (node != null) {
                throw new RRException("当前任务节点名称已存在");
            }
            this.flinkTaskNodeMapper.insertOrUpdateSelective(taskNode);
        }
        if (vo.getConfig() != null) {
            String config = vo.getConfig().toJSONString();
            realTimeTaskMapper.updateConfig(config, vo.getTaskId());
        }
        if (CollectionUtils.isNotEmpty(vo.getFieldList())) {
            for (FlinkNodeField flinkNodeField : vo.getFieldList()) {
                flinkNodeField.setNodeId(taskNode.getId());
                flinkNodeField.preSave(userId);
            }
            flinkNodeFieldMapper.delete(new QueryWrapper<FlinkNodeField>().eq("node_id", taskNode.getId()));
            flinkNodeFieldMapper.batchInsert(vo.getFieldList());
        }
        vo.setId(taskNode.getId());
    }

    @Override
    public FlinkTaskNodeVO getDetailById(Integer id) {
        FlinkTaskNode node = flinkTaskNodeMapper.selectById(id);
        FlinkTaskNodeVO vo = new FlinkTaskNodeVO(node);
        if (node != null) {
            List<FlinkNodeField> list = flinkNodeFieldMapper.selectList(new QueryWrapper<FlinkNodeField>().eq("node_id", id));
            if (CollectionUtils.isNotEmpty(list)) {
                vo.setFieldList(list);
            }
        }
        return vo;
    }
}
